package com.atguigu.gmall.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.BaseAppV1;
import com.atguigu.gmall.realtime.bean.TableProcess;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.util.FlinkSinkUtil;
import com.atguigu.gmall.realtime.util.JdbcUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Properties;

/**
 * @Author lzc
 * @Date 2022/12/2 11:14
 */
@Slf4j
public class Dwd_08_BaseDb extends BaseAppV1 {
    public static void main(String[] args) {
        new Dwd_08_BaseDb().init(
            3008,
            2,
            "Dwd_08_BaseDb",
            Constant.TOPIC_ODS_DB
        );
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          DataStreamSource<String> stream) {
        // 实现业务: 对流做各种操作
        // 1. 清洗数据
        SingleOutputStreamOperator<JSONObject> etledStream = etl(stream);
        // 2. 读取配置信息
        SingleOutputStreamOperator<TableProcess> tpStream = readTableProcess(env);
        // 4. 过滤出需要的数据, 分流(动态分流技术 cdc connect 广播状态 )
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> dimToConfigStream = connect(etledStream, tpStream);
        // 5. 删除不需要的列
        dimToConfigStream = delNotNeedColumns(dimToConfigStream);
        // 6. 写出到到 kafka 中
        writeToKafka(dimToConfigStream);
        
        
    }
    
    private void writeToKafka(SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> dimToConfigStream) {
        dimToConfigStream.addSink(FlinkSinkUtil.getKafkaSink());
    }
    
    private void writeToPhoenix(SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> stream) {
        /*
        如何写出到 phoenix 中?
        1. 找到 phoenix 连接器. 没有
        2. 使用 jdbc 连接器?
            不能. 因为 jdbc sink 只能把流中的数据写入到一个表中
            
        3. 只能 自定义 sink
         */
        stream.addSink(FlinkSinkUtil.getPhoenixSink());
        
    }
    
    private SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> delNotNeedColumns(
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> stream) {
        return stream.map(new MapFunction<Tuple2<JSONObject, TableProcess>, Tuple2<JSONObject, TableProcess>>() {
            @Override
            public Tuple2<JSONObject, TableProcess> map(Tuple2<JSONObject, TableProcess> t) throws Exception {
                //                System.out.println(t);
                JSONObject data = t.f0;  // 本质就是 map 集合, 里面的一些 key 没有存在必要,要删除
                List<String> columns = Arrays.asList(t.f1.getSinkColumns().split(","));
                data.keySet().removeIf(key -> !columns.contains(key) && !"op_type".equals(key));
                
                return t;
            }
        });
    }
    
    // 连接数据流和配置流
    private SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> connect(
        SingleOutputStreamOperator<JSONObject> dataStream,
        SingleOutputStreamOperator<TableProcess> tpStream) {
        MapStateDescriptor<String, TableProcess> tpStateDesc =
            new MapStateDescriptor<>("tpState", String.class, TableProcess.class);
        BroadcastStream<TableProcess> tpBcStream = tpStream.broadcast(tpStateDesc);
        // 2. 让数据流去 connect 广播流
        return dataStream
            .connect(tpBcStream)
            .process(new BroadcastProcessFunction<JSONObject, TableProcess, Tuple2<JSONObject, TableProcess>>() {
                
                private HashMap<String, TableProcess> tpMap;
                private Connection conn;
                
                @Override
                public void open(Configuration parameters) throws Exception {
                    // 在这里提前把所有的配置信息都进来
                    // 使用基本的 jdbc 连接读取 mysql 中的配置数据
                    // 读取到的数据存储到什么位置?  在 flink 中的, 所有的状态的操作,都无法在 open 中使用
                    // 使用一个普通的 HashMap 就行了
                    conn = JdbcUtil.getMysqlConnection();
                    /*
                        id  name    age
                        1    lisi    10
                        2    zs      20
                     */
                    String sql = "select * from table_process";
                    List<TableProcess> tps = JdbcUtil.queryList(conn, sql, null, TableProcess.class, true);
                    
                    tpMap = new HashMap<>();
                    for (TableProcess tp : tps) {
                        String key = tp.getSourceTable() + ":" + tp.getSourceType()
                            + (tp.getSinkExtend() == null ? "" : tp.getSinkExtend());
                        tpMap.put(key, tp);
                        log.warn("初始化配置表: " + tp);
                    }
                }
                
                @Override
                public void close() throws Exception {
                    JdbcUtil.closeConnection(conn);
                }
                
                // 处理数据流中的数据: 维度数据
                @Override
                public void processElement(JSONObject obj,
                                           ReadOnlyContext ctx,
                                           Collector<Tuple2<JSONObject, TableProcess>> out) throws Exception {
                    // 4. 处理数据流中的数据, 从广播状态读取数据
                    ReadOnlyBroadcastState<String, TableProcess> state = ctx.getBroadcastState(tpStateDesc);
                    String table = obj.getString("table");
                    String type = obj.getString("type");
                    String key = table + ":" + type;
                    
                    // 特殊性情况: 当是优惠券的下单和支付的时候, 需要特殊对待
                    if ("coupon_use".equals(table) && "update".equals(type)) {
                        String oldStatus = obj.getJSONObject("old").getString("coupon_status");
                        String newStatus = obj.getJSONObject("data").getString("coupon_status");
                        String dataUsedTime = obj.getJSONObject("data").getString("used_time");
                        if ("1401".equals(oldStatus) && "1402".equals(newStatus)) {
                            key += "{\"data\": {\"coupon_status\": \"1402\"}, \"old\": {\"coupon_status\": \"1401\"}}";
                        } else if (null != dataUsedTime) {
                            key += "{\"data\": {\"used_time\": \"not null\"}}";
                        }
                    }
                    /*
                    tpMap中的配置信息, 是在程序启动的时候初始化的.
                    先从广播状态中读取, 如果没有读到,再从 map 中读取
                     */
                    TableProcess tp = state.get(key);
                    if (tp == null) {
                        tp = tpMap.get(key);
                        if (tp != null) {
                            log.warn("从 map 中读取配置信息...");
                        }
                    }
                    
                    if (tp != null) {
                        JSONObject data = obj.getJSONObject("data");
                        data.put("op_type", obj.getString("type")); // 这个字段后面要用
                        out.collect(Tuple2.of(data, tp));
                    }
                }
                
                // 处理广播流中的数据: 配置信息
                @Override
                public void processBroadcastElement(TableProcess tp,
                                                    Context ctx,
                                                    Collector<Tuple2<JSONObject, TableProcess>> out) throws Exception {
                    
                    // 3. 把配置信息放入到广播状态中
                    BroadcastState<String, TableProcess> state = ctx.getBroadcastState(tpStateDesc);
                    String key = tp.getSourceTable() + ":" + tp.getSourceType() + (tp.getSinkExtend() == null ? "" : tp.getSinkExtend());
                    // 不同的配置信息的操作, 对状态的操作应该不同
                    if ("d".equals(tp.getOp())) { // 配置信息删除了一条配置
                        state.remove(key);
                        // 预加载的配置信息也应该删除
                        tpMap.remove(key);
                    } else {
                        state.put(key, tp);
                    }
                    
                }
            });
        
        
    }
    
    
    private SingleOutputStreamOperator<TableProcess> readTableProcess(StreamExecutionEnvironment env) {
        Properties props = new Properties();
        props.setProperty("useSSL", "false");
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
            .hostname("hadoop162")
            .port(3306)
            .jdbcProperties(props)
            .databaseList("gmall_config") // set captured database, If you need to synchronize the whole database, Please set tableList to ".*".
            .tableList("gmall_config.table_process") // set captured table
            .username("root")
            .password("aaaaaa")
            .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
            .build();
        
        return env
            .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "mysql-source")
            .map(new MapFunction<String, TableProcess>() {
                @Override
                public TableProcess map(String json) throws Exception {
                    JSONObject obj = JSON.parseObject(json);
                    String op = obj.getString("op");
                    TableProcess tp;
                    if (!"d".equals(op)) {
                        tp = obj.getObject("after", TableProcess.class);
                    } else {
                        tp = obj.getObject("before", TableProcess.class);
                    }
                    
                    tp.setOp(op);
                    return tp;
                }
            })
            .filter(tp -> "dwd".equals(tp.getSinkType()));
        
        
    }
    
    private SingleOutputStreamOperator<JSONObject> etl(DataStreamSource<String> stream) {
        return stream
            .filter(new FilterFunction<String>() {
                @Override
                public boolean filter(String value) throws Exception {
                    // 1. value是 json 格式
                    // 2. 数据库必须满足
                    try {
                        JSONObject obj = JSON.parseObject(value);
                        
                        String type = obj.getString("type");
                        
                        String data = obj.getString("data");
                        
                        return "gmall2022".equals(obj.getString("database"))
                            && ("insert".equals(type) || "update".equals(type) || "bootstrap-insert".equals(type))
                            && data != null
                            && data.length() > 2;
                    } catch (Exception e) {
                        log.warn("数据格式有误,不是正确的 json 数据: " + value);
                        return false;
                    }
                }
            })
            .map(json -> JSON.parseObject(json.replaceAll("bootstrap-", "")));
        
    }
}
/*
对维度数据来说, 很多是已经存在很久了.

分析的业务的时候, 用到旧的维度信息, 维度需要同步历史数据!!!

maxwell 能不能同步历史数据? 能
maxwell-bootstrap
--------------------
https://developer.aliyun.com/article/777502
过滤后的数据中, 既有维度表的数据, 又有事实表的数据

把需要的维度数据写出到 phoenix 中

如何识别出来需要的维度数据?
 1. 在代码中, 硬编码每张维度表. 不可取
 
 2. 把需要的维度表的信息, 放入配置表中, flink 程序实时的读取配置表的数据.

如何让配置表实时的控制维度数据的处理?
  ods_db来的维度数据: 数据流
  配置流: 做成广播流
  数据流去 connect 广播流
  
flink cdc: 直接从 mysql 读取数据(变化数据)

----

读取的配置数据中 op:
r: 当程序启动的时候, 读取的快照   before: null after: 有\
    建表
u: 更新某个字段的时候  before:有  after: 有
    先删表,再建表
c: 新增数据 before: null after: 有
    建表
d: 删除数据 before: 有 after: null
    删表

如果修改的是主键: 则先 d 再 c
----------

数据库.表名

phoenix 也可以有数据库
hbase 有没有数据库? 没有. 有一个东西 namespace

<property>
    <name>phoenix.schema.isNamespaceMappingEnabled</name>
    <value>true</value>
</property>

<property>
    <name>phoenix.schema.mapSystemTablesToNamespace</name>
    <value>true</value>
</property>
服务器开了映射, 则客户端也必须开
-----
 SALT_BUCKETS = 4  建盐表
 
 reginServer
 
 regin 存在与reginServer
 
 每间一张表, 默认只有一个 region
 
 当 region 变的越来愈大的时候, 会自动分裂
 0.93 之前: 10G 一分为二
 0.93 开始: 2*n^3*128 开始分裂
 
 自动迁移: 当集群的负载不大迁移
 
 为了避免分裂和迁移: 建预分区表
 
    a|  b|
 
 -------
 盐表: 对应着 hbase 中的预分区表
 
 0x01 0x02 0x03 0x04
 
 
 bin/flink run-application -d
 -t yarn-application
 -c com.atguigu.gmall.realtime.app.dim.DimApp
 -Dclassloader.check-leaked-classloader=false
 -Dclassloader.resolve-order=parent-first
 /opt/gmall220704/gmall-realtime-1.0-SNAPSHOT.jar
 
 */









